import rx.Observable
import rx.Subscriber
import rx.internal.producers.SingleDelayedProducer

/**
 * 打包发送到下游
 * @note toList有同样的效果
 */

fun <T> Observable<T>.pack(): Observable<List<T>> {
    return this.lift(PackOperator<T>())
}

open class PackOperator<T> : Observable.Operator<List<T>, T> {
    override fun call(t: Subscriber<in List<T>>): Subscriber<in T> {
        val bowl = ArrayList<T>()
        val producer = SingleDelayedProducer<List<T>>(t)
        val s = object : Subscriber<T>() {
            override fun onStart() {
                request(Long.MAX_VALUE)
            }

            override fun onNext(t: T) {
                bowl.add(t)
            }

            override fun onCompleted() {
                producer.setValue(bowl)
            }

            override fun onError(e: Throwable?) {
                t.onError(e)
            }

        }
        t.add(s)
        t.setProducer(producer)
        return s
    }


}

fun main(args: Array<String>) {

    var observable = Observable
            .create(Observable.OnSubscribe<Int> {
                for (k in 0..10)
                    it.onNext(k)
                it.onCompleted()
            })
    observable.pack()
            .subscribe({
                println(it)
            })

    //toList的等效替换(哈哈，其实内部实现也是使用了SingleDelayProducer)
    observable.toList().subscribe({
        println(it)
    })


}